
                /*********************************************
                **                                          **
                **                                          **
                **              ca.prb.cml.C                **
                **                                          **
                **                                          **
                *********************************************/
                /*********************************************
                /*********************************************
                **                                          **
                **         Author Jean-Marc Adamo           **
                **       Copyright (C) 1997-2008 JMA        **
                *********************************************/

/*
Asynchronous Communication
Check probe functions.
Needs exactly 3 processors to execute.
      ^^^^^^^^^
*/

#include "arch.h"

Global_id gid1;
Global_id gid2;
Global_id gid3;
Global_id gid4;

Coll_id cid1, cid2, cid3;

void init_gid(int, char **){

   gid1.set();
   gid2.set();
   gid3.set();
   gid4.set();

   cid1.set(),
   cid2.set(),
   cid3.set(),
   fprintf(Sdf::trace, "node%d: gid1=%d, gid2=%d, gid3=%d, gid4=%d, cid1=%d, cid2=%d, cid3=%d\n",
                Sdf::PROC_ID, 
                gid1.get(), gid2.get(), 
                gid3.get(),gid4.get(),
                cid1.get(), cid2.get(), cid3.get());
   fflush(Sdf::trace);
}

class SR_count{
   public:
      int cnt;
      SR_count(){
         cnt = 0;
      }
      void reset(){
         cnt = 0;
      }
      void more_SR(int i){
         cnt += i;
      }
};

#define wait_while(condition) while(condition) Sdf::poll();
#define not_complete  (c.cnt >0)

ChanA<int> *chanl;
ChanA<int> *chanli;
ChanA<int> *chanls;
ChanA<int> *chanld;

void upon_end_of_SR_r(ChanAs *chan){

   fprintf(Sdf::trace, "node%d, r>>>>>>>\n", Sdf::PROC_ID);
   Sdf::Access_type access_type;
   int tag_dv = chan->get_tag_dv(&access_type);
   fprintf(Sdf::trace, "node%d, upon_end_of_SR_r, capa=%d, rcvd size=%d, Tsize=%d, rps=%d, rpr=%d, tag_dv=%d, access_type=%d, tag_av=%d, tag_ev=%d, comm=%d\n", 
                Sdf::PROC_ID, 
                chan->get_chan_capacity(), 
                chan->get_msg_size(Sdf::RCV), 
                chan->get_Tsize(),
                chan->get_remote_proc(Sdf::SND), 
                chan->get_remote_proc(Sdf::RCV),
                tag_dv,
                access_type,
                chan->get_tag_av(),
                chan->get_tag_ev(),
                chan->get_comm());

   ((SR_count *)(chan->get_hndlr_data_ptr(Sdf::RCV)))->cnt--;

   fprintf(Sdf::trace, "node%d, r<<<<<<<\n", Sdf::PROC_ID);
   fflush(Sdf::trace);
}

void upon_end_of_SR_s(ChanAs *chan){

   fprintf(Sdf::trace, "node%d, s>>>>>>>\n", Sdf::PROC_ID);
   Sdf::Access_type access_type;
   int tag_dv = chan->get_tag_dv(&access_type);
   fprintf(Sdf::trace, "node%d, upon_end_of_SR_s, capa=%d, sent size=%d, Tsize=%d, rps=%d, rpr=%d, tag_dv=%d, access_type=%d, tag_av=%d, tag_ev=%d, comm=%d\n", 
            Sdf::PROC_ID, 
                chan->get_chan_capacity(), 
                chan->get_msg_size(Sdf::SND), 
                chan->get_Tsize(),
                chan->get_remote_proc(Sdf::SND), 
                chan->get_remote_proc(Sdf::RCV),
                tag_dv,
                access_type,
                chan->get_tag_av(),
                chan->get_tag_ev(),
                chan->get_comm());

   ((SR_count *)(chan->get_hndlr_data_ptr(Sdf::SND)))->cnt--;

   fprintf(Sdf::trace, "node%d, s<<<<<<<\n", Sdf::PROC_ID);
   fflush(Sdf::trace);
}

class  Root_thread: public Thread{
   int recvd;
   public:
        Root_thread()
                : Thread((Tmf0_ptr)&Root_thread::body){}

        void body(){

            fprintf(Sdf::trace, "node%d, start\n", Sdf::PROC_ID);

            int comml_rank;
            int comml_size;

            int commli_rank;
            int commli_size;

            int commls_rank;
            int commls_size;

            int commld_rank;
            int commld_size;

            CommL *comml;
               //commli will be defined as identical to comml
            CommL *commli;
               //commls will be defined as similar to comml
            CommL *commls;
               //commld will be defined as different from comml
            CommL *commld;

            int colour;
            if(Sdf::PROC_ID==2)
               colour=Sdf::UNDEFINED;
            else
               colour=1;
            CommL **comml_table = 
            Sdf::COMML_WORLD->split(cid1, 1, colour, Sdf::PROC_ID);
            comml = *comml_table;

            int ranks[2] = {0, 1};
            Group *groupli = Sdf::incl(comml, 2, ranks);
            commli = groupli->commit(cid1, Sdf::COMML_WORLD);

            CommL **commls_table = 
            Sdf::COMML_WORLD->split(cid1, 1, colour, 1-Sdf::PROC_ID);
            commls = *commls_table;

            if(Sdf::PROC_ID==0)
               colour=Sdf::UNDEFINED;
            else
               colour=1;
            CommL **commld_table = 
            Sdf::COMML_WORLD->split(cid1, 1, colour, 2-Sdf::PROC_ID);
            commld = *commld_table;

            comml_rank = comml->rank();
            comml_size = comml->size();

            commli_rank = commli->rank();
            commli_size = commli->size();

            commls_rank = commls->rank();
            commls_size = commls->size();

            commld_rank = commld->rank();
            commld_size = commld->size();

            fprintf(Sdf::trace, "node%d, ranks & sizes, %d, %d; %d, %d; %d, %d, %d, %d\n", 
                    Sdf::PROC_ID,
                    comml_rank,  comml_size,
                    commli_rank, commli_size,
                    commls_rank, commls_size,
                    commld_rank, commld_size);
            fflush(Sdf::trace);

            SR_count c;

            if(Sdf::PROC_ID!=2){

               chanl  = new ChanA<int>(gid1, (1-comml_rank),
                                       upon_end_of_SR_s, &c,
                                       upon_end_of_SR_r, &c,
                                       1,
                                       comml);

               chanli = new ChanA<int>(gid2, (1-commli_rank),
                                       upon_end_of_SR_s, &c,
                                       upon_end_of_SR_r, &c,
                                       1,
                                       commli);

               chanls = new ChanA<int>(gid3, (1-commls_rank),
                                       upon_end_of_SR_s, &c,
                                       upon_end_of_SR_r, &c,
                                       1,
                                       commls);
            }

            if(Sdf::PROC_ID != 0){

               chanld  = new ChanA<int>(gid4, (1-commld_rank), 
                                       upon_end_of_SR_s, &c,
                                       upon_end_of_SR_r, &c,
                                       1,
                                       commld); 
            }

            fprintf(Sdf::trace, "node%d, before Sdf::barrier\n", Sdf::PROC_ID);
            fflush(Sdf::trace);

            Sdf::barrier(cid2);

            fprintf(Sdf::trace, "node%d, after Sdf::barrier\n", Sdf::PROC_ID);
            fflush(Sdf::trace);

            if(!Sdf::PROC_ID){
               c.more_SR(3);
               chanls->send(Sdf::STD, 0);
               chanli->send(Sdf::STD, 1);
               chanl->send(Sdf::STD,  2);
               fprintf(Sdf::trace, "node%d, after send sequence\n",
                       Sdf::PROC_ID);
               fflush(Sdf::trace);
               wait_while(not_complete);
               fprintf(Sdf::trace, "node%d, after completion\n",
                       Sdf::PROC_ID);
               fflush(Sdf::trace);
            }else{
               if(Sdf::PROC_ID==2){
                  c.more_SR(1);
                  chanld->send(Sdf::STD, 10);
                  fprintf(Sdf::trace, "node%d, after send\n",
                          Sdf::PROC_ID);
                  fflush(Sdf::trace);
                  wait_while(not_complete);
                  fprintf(Sdf::trace, "node%d, after completion\n",
                          Sdf::PROC_ID);
                  fflush(Sdf::trace);
               }else{ //Sdf::PROC_ID==1
                  int recvd;
                    //chanl is first probed and read
                  while(1){
                     int tag = chanl->probe(); 
                     if(tag == chanl->get_tag_av()){
                        c.more_SR(1);
                        chanl->recv(recvd); 
                        wait_while(not_complete);
                        fprintf(Sdf::trace, "node%d, chanl.probe(), tag=%d, recvd=%d\n",
                                Sdf::PROC_ID, tag, recvd);
                        fflush(Sdf::trace);
                        break;
                     }else{
                        if(tag){
                           fprintf(Sdf::trace, "node%d, something wrong happened 1, tag=%d\n",
                                   Sdf::PROC_ID, tag);
                           fflush(Sdf::trace);
                        }
                     }
                  }
                    //probing for msgs on ChanAs based on CommL
                    //groups identical to comml (i.e. commli).
                  while(1){
                     int tag = Sdf::ANY.probe(comml, Sdf::IDENT);
                     if(tag == chanli->get_tag_av()){
                        c.more_SR(1);
                        chanli->recv(recvd);
                        wait_while(not_complete);
                        fprintf(Sdf::trace, "node%d, Sdf::ANY.probe(comml, Sdf::IDENT), tag=%d, recvd=%d\n",
                                Sdf::PROC_ID, tag, recvd);
                        fflush(Sdf::trace);
                        break;
                     }else{
                        if(tag){
                           fprintf(Sdf::trace, "node%d, something wrong happened 2, tag=%d\n",
                                   Sdf::PROC_ID, tag);
                           fflush(Sdf::trace);
                        }
                     }
                  }
                    //probing for msgs on ChanAs based on CommL
                    //groups similar to comml (i.e. commls).
                  while(1){
                     int tag = Sdf::ANY.probe(comml, Sdf::SIMLR);
                     if(tag == chanls->get_tag_av()){
                        c.more_SR(1);
                        chanls->recv(recvd);
                        wait_while(not_complete);
                        fprintf(Sdf::trace, "node%d, Sdf::ANY.probe(comml, Sdf::SIMILR), tag=%d, recvd=%d\n",
                                Sdf::PROC_ID, tag, recvd);
                        fflush(Sdf::trace);
                        break;
                     }else{
                        if(tag){
                           fprintf(Sdf::trace, "node%d, something wrong happened 3, tag=%d\n",
                                   Sdf::PROC_ID, tag);
                           fflush(Sdf::trace);
                        }
                     }
                  }
                    //probing for msgs on ChanAs based on CommL
                    //groups different from comml (i.e. commls).
                  while(1){
                     int tag = Sdf::ANY.probe(comml, Sdf::UNEQL);
                     if(tag == chanld->get_tag_av()){
                        c.more_SR(1);
                        chanld->recv(recvd);
                        wait_while(not_complete);
                        fprintf(Sdf::trace, "node%d, Sdf::ANY.probe(comml, Sdf::UNEQL), tag=%d, recvd=%d\n",
                                Sdf::PROC_ID, tag, recvd);
                        fflush(Sdf::trace);
                        break;
                     }else{
                        if(tag){
                           fprintf(Sdf::trace, "node%d, something wrong happened 4, tag=%d\n",
                                   Sdf::PROC_ID, tag);
                           fflush(Sdf::trace);
                        }
                     }
                  }
               }
            }

            Sdf::barrier(cid3);
              //this is to protect channel destruction
            delete chanl;
            delete chanli;
            delete chanls;
            delete chanld;

            delete comml;
            delete commli;
            delete commls;
            delete commld;
       }
};

MAIN(Root_thread, init_gid, Sdf::empty, Sdf::empty)

